﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using System.Data.SqlClient;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Data;

namespace ssisCTC
{
    [DtsPipelineComponent(DisplayName = "BatchDestination",
        ComponentType = ComponentType.DestinationAdapter,
        IconResource = "ssisCTC.BatchDestination.ico",
        UITypeName = "ssisCTC.BatchDestinationUI, ssisCTC.BatchDestination, Version=1.0.0.0, Culture=neutral, PublicKeyToken=41aa3e6898abe909")]
    public class BatchDestination : PipelineComponent
    {
        private SqlConnection _conn = null;
        private List<ColumnInfo> _columnInfos = null;
        private DataTable _rowsToWrite = null;

        #region Design Time

        public override void ProvideComponentProperties()
        {
            base.ProvideComponentProperties();

            //Clear out base implmentation
            this.ComponentMetaData.RuntimeConnectionCollection.RemoveAll();
            this.ComponentMetaData.InputCollection.RemoveAll();
            this.ComponentMetaData.OutputCollection.RemoveAll();

            //Add a single input
            IDTSInput100 input = this.ComponentMetaData.InputCollection.New();
            input.Name = "Batch Input";
            input.Description = "Input for the Batch Destination";
            input.HasSideEffects = true;

            //No outputs - not handling errors currently

            //New connection manager
            IDTSRuntimeConnection100 conn = this.ComponentMetaData.RuntimeConnectionCollection.New();
            conn.Name = Constants.CONNECTION_MANAGER_NAME;
            conn.Description = "ADO.NET Connection to SQL Server";

            //Add a property to store the command to execute
            AddProperty(Constants.PROP_SQL_COMMAND, "Command to execute against the batch", string.Empty, true);

            //Add a property to store the table name to use
            AddProperty(Constants.PROP_WORKING_TABLE, "Working table to store temporary tables", this.ComponentMetaData.Name, true);
        }

        public override void ReinitializeMetaData()
        {
            base.ReinitializeMetaData();
        }

        #endregion

        #region Run Time

        public override void AcquireConnections(object transaction)
        {
            IDTSRuntimeConnection100 runtimeConn = this.ComponentMetaData.RuntimeConnectionCollection[Constants.CONNECTION_MANAGER_NAME];
            if (runtimeConn == null || runtimeConn.ConnectionManager == null)
            {
                FireEvent(EventType.Error, "The Connection Manager must be set.");
            }
            else
            {
                object tempConn = this.ComponentMetaData.RuntimeConnectionCollection[Constants.CONNECTION_MANAGER_NAME].ConnectionManager.AcquireConnection(transaction);
                if (tempConn is SqlConnection)
                {
                    _conn = (SqlConnection)tempConn;
                    if (_conn.State == System.Data.ConnectionState.Closed)
                    {
                        _conn.Open();
                    }
                }
                else
                {
                    FireEvent(EventType.Error, "The BatchDestination can only use ADO.NET SQL Server connections.");
                }
            }
        }

        public override DTSValidationStatus Validate()
        {
            if (this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_SQL_COMMAND].Value == null ||
                this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_SQL_COMMAND].Value.ToString() == string.Empty)
            {
                FireEvent(EventType.Error, "The SQLCommand property must be set.");
                return DTSValidationStatus.VS_ISBROKEN;
            }
            if (this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_WORKING_TABLE].Value.ToString() == string.Empty)
            {
                FireEvent(EventType.Error, "The WorkingTable property must be set.");
                return DTSValidationStatus.VS_ISBROKEN;
            }
            //TODO: Add Validation of SQL Command - syntax check


            return base.Validate();
        }

        public override void ReleaseConnections()
        {
            base.ReleaseConnections();
            if (_conn != null && _conn.State != System.Data.ConnectionState.Closed)
            {
                _conn.Close();
            }
        }

        public override void PreExecute()
        {
            _columnInfos = new List<ColumnInfo>(this.ComponentMetaData.InputCollection[0].InputColumnCollection.Count);

            IDTSInput100 input = this.ComponentMetaData.InputCollection[0];
            // Look at each input, and then each column, storing important metadata.
            foreach (IDTSInputColumn100 col in input.InputColumnCollection)
            {
                // Find the position in buffers that this column will take, and add it to the map.
                _columnInfos.Add(new ColumnInfo(col.Name, col.DataType, this.BufferManager.FindColumnByLineageID(input.Buffer, col.LineageID), col.Length, col.Precision, col.Scale));
            }
            try
            {
                //Create Working Table
                string sqlText = "CREATE TABLE " + (string)this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_WORKING_TABLE].Value + Environment.NewLine;
                sqlText += "( " + GetColumnSQL(_columnInfos) + " )";
                SqlCommand comm = _conn.CreateCommand();
                comm.CommandText = sqlText;
                comm.ExecuteNonQuery();

                _rowsToWrite = new DataTable((string)this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_WORKING_TABLE].Value);

                foreach (ColumnInfo col in _columnInfos)
                {
                    _rowsToWrite.Columns.Add(col.ColumnName, GetNetDataType(col));
                }
            }
            catch (Exception ex)
            {
                FireEvent(EventType.Error, ex.Message);
                throw ex;
            }

        }

        public override void ProcessInput(int inputID, PipelineBuffer buffer)
        {
            base.ProcessInput(inputID, buffer);
            while (buffer.NextRow())
            {
                object[] values = new object[_columnInfos.Count];
                int counter = 0;

                foreach (ColumnInfo col in _columnInfos)
                {
                    values[counter] = GetBufferColumnValue(buffer, col);
                    counter++;
                }
                _rowsToWrite.Rows.Add(values);
            }

            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(_conn))
            {
                bulkCopy.DestinationTableName = this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_WORKING_TABLE].Value.ToString();

                try
                {
                    // Write from the source to the destination.
                    bulkCopy.WriteToServer(_rowsToWrite);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
            }

            this.ComponentMetaData.IncrementPipelinePerfCounter(Constants.DTS_PIPELINE_CTR_ROWSWRITTEN, (uint)_rowsToWrite.Rows.Count);
            _rowsToWrite.Rows.Clear();
        }

        public override void PostExecute()
        {
            base.PostExecute();

            try
            {
                int rowsAffected = 0;
                string sqlText = (string)this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_SQL_COMMAND].Value;
                SqlCommand comm = _conn.CreateCommand();
                comm.CommandText = sqlText;
                rowsAffected = comm.ExecuteNonQuery();
                FireEvent(EventType.Information, rowsAffected.ToString() + " rows were affected by the batch SQL Command.");
            }
            catch (Exception ex)
            {
                FireEvent(EventType.Error, ex.Message);
            }

            try
            {
                string sqlText = "DROP TABLE " + (string)this.ComponentMetaData.CustomPropertyCollection[Constants.PROP_WORKING_TABLE].Value + Environment.NewLine;
                SqlCommand comm = _conn.CreateCommand();
                comm.CommandText = sqlText;
                //comm.ExecuteNonQuery();
            }
            catch (Exception ex)
            {
                FireEvent(EventType.Error, ex.Message);
            }
        }

        #endregion

        #region Helpers

        private static object GetBufferColumnValue(PipelineBuffer buffer, ColumnInfo col)
        {
            if (buffer.IsNull(col.BufferIndex))
            {
                return null;
            }
            switch (col.ColumnDataType)
            {
                case DataType.DT_BOOL:
                    return buffer.GetBoolean(col.BufferIndex);
                case DataType.DT_BYTES:
                    return buffer.GetBytes(col.BufferIndex);
                case DataType.DT_CY:
                    return buffer.GetDecimal(col.BufferIndex);
                case DataType.DT_DATE:
                    return buffer.GetDateTime(col.BufferIndex);
                case DataType.DT_DBDATE:
                    return buffer.GetDate(col.BufferIndex);
                case DataType.DT_DBTIME:
                    return buffer.GetTime(col.BufferIndex);
                case DataType.DT_DBTIME2:
                    return buffer.GetTime(col.BufferIndex);
                case DataType.DT_DBTIMESTAMP:
                    return buffer.GetDateTime(col.BufferIndex);
                case DataType.DT_DBTIMESTAMP2:
                    return buffer.GetDateTime(col.BufferIndex);
                case DataType.DT_DBTIMESTAMPOFFSET:
                    return buffer.GetDateTimeOffset(col.BufferIndex);
                case DataType.DT_DECIMAL:
                    return buffer.GetDecimal(col.BufferIndex);
                case DataType.DT_FILETIME:
                    return buffer.GetDateTime(col.BufferIndex);
                case DataType.DT_GUID:
                    return buffer.GetGuid(col.BufferIndex);
                case DataType.DT_I1:
                    return buffer.GetSByte(col.BufferIndex);
                case DataType.DT_I2:
                    return buffer.GetInt16(col.BufferIndex);
                case DataType.DT_I4:
                    return buffer.GetInt32(col.BufferIndex);
                case DataType.DT_I8:
                    return buffer.GetInt64(col.BufferIndex);
                case DataType.DT_IMAGE:
                    return buffer.GetBlobData(col.BufferIndex, 0, (int)buffer.GetBlobLength(col.BufferIndex));
                case DataType.DT_NTEXT:
                    return buffer.GetBlobData(col.BufferIndex, 0, (int)buffer.GetBlobLength(col.BufferIndex));
                case DataType.DT_NUMERIC:
                    return buffer.GetDecimal(col.BufferIndex);
                case DataType.DT_R4:
                    return buffer.GetSingle(col.BufferIndex);
                case DataType.DT_R8:
                    return buffer.GetDouble(col.BufferIndex);
                case DataType.DT_STR:
                    return buffer.GetString(col.BufferIndex);
                case DataType.DT_TEXT:
                    return buffer.GetBlobData(col.BufferIndex, 0, (int)buffer.GetBlobLength(col.BufferIndex));
                case DataType.DT_UI1:
                    return buffer.GetByte(col.BufferIndex);
                case DataType.DT_UI2:
                    return buffer.GetUInt16(col.BufferIndex);
                case DataType.DT_UI4:
                    return buffer.GetUInt32(col.BufferIndex);
                case DataType.DT_UI8:
                    return buffer.GetUInt64(col.BufferIndex);
                case DataType.DT_WSTR:
                    return buffer.GetString(col.BufferIndex);
                default:
                    return null;
            }
        }

        private string GetColumnSQL(List<ColumnInfo> columnInfos)
        {
            string returnValue = string.Empty;
            int counter = 0;

            foreach (ColumnInfo col in columnInfos)
            {
                if (counter > 0) returnValue += ", ";
                returnValue += "[" + col.ColumnName + "] " + GetSQLServerDataType(col) + Environment.NewLine;
                counter++;
            }

            return returnValue;
        }

        private static string GetSQLServerDataType(ColumnInfo col)
        {
            switch (col.ColumnDataType)
            {
                case DataType.DT_BOOL:
                    return "bit";
                case DataType.DT_BYTES:
                    return "varbinary(" + col.Length.ToString() + ")";
                case DataType.DT_CY:
                    return "money";
                case DataType.DT_DATE:
                    return "date";
                case DataType.DT_DBDATE:
                    return "date";
                case DataType.DT_DBTIME:
                    return "time";
                case DataType.DT_DBTIME2:
                    return "time";
                case DataType.DT_DBTIMESTAMP:
                    return "datetime";
                case DataType.DT_DBTIMESTAMP2:
                    return "datetime2";
                case DataType.DT_DBTIMESTAMPOFFSET:
                    return "datetimeoffset";
                case DataType.DT_DECIMAL:
                    return "numeric(" + col.Precision.ToString() + "," + col.Scale.ToString() + ")";
                case DataType.DT_GUID:
                    return "uniqueidentifier";
                case DataType.DT_I1:
                    return "smallint";
                case DataType.DT_I2:
                    return "smallint";
                case DataType.DT_I4:
                    return "int";
                case DataType.DT_I8:
                    return "bigint";
                case DataType.DT_IMAGE:
                    return "image";
                case DataType.DT_NTEXT:
                    return "ntext";
                case DataType.DT_NUMERIC:
                    return "numeric(" + col.Precision.ToString() + "," + col.Scale.ToString() + ")";
                case DataType.DT_R4:
                    return "real";
                case DataType.DT_R8:
                    return "real";
                case DataType.DT_STR:
                    return "varchar(" + col.Length.ToString() + ")";
                case DataType.DT_TEXT:
                    return "text";
                case DataType.DT_UI1:
                    return "tinyint";
                case DataType.DT_UI2:
                    return "int";
                case DataType.DT_UI4:
                    return "bigint";
                case DataType.DT_UI8:
                    return "bigint";
                case DataType.DT_WSTR:
                    return "nvarchar(" + col.Length.ToString() + ")";
                default:
                    return string.Empty;
            }
        }

        private static System.Type GetNetDataType(ColumnInfo col)
        {
            switch (col.ColumnDataType)
            {
                case DataType.DT_BOOL:
                    return typeof(bool);

                case DataType.DT_IMAGE:
                case DataType.DT_BYTES:
                    return typeof(byte);

                case DataType.DT_NUMERIC:
                case DataType.DT_DECIMAL:
                case DataType.DT_CY:
                    return typeof(decimal);

                case DataType.DT_DATE:
                case DataType.DT_DBDATE:
                case DataType.DT_DBTIME:
                case DataType.DT_DBTIME2:
                case DataType.DT_DBTIMESTAMP:
                case DataType.DT_DBTIMESTAMP2:
                    return typeof(DateTime);

                case DataType.DT_DBTIMESTAMPOFFSET:
                    return typeof(DateTimeOffset);
                case DataType.DT_GUID:
                    return typeof(Guid);

                case DataType.DT_I1:
                case DataType.DT_I2:
                    return typeof(Int16);

                case DataType.DT_I4:
                    return typeof(Int32);
                case DataType.DT_I8:
                    return typeof(Int64);


                case DataType.DT_R4:
                case DataType.DT_R8:
                    return typeof(Single);

                case DataType.DT_UI1:
                case DataType.DT_UI2:
                case DataType.DT_UI4:
                    return typeof(uint);

                case DataType.DT_UI8:
                    return typeof(UInt64);

                case DataType.DT_WSTR:
                case DataType.DT_STR:
                case DataType.DT_TEXT:
                case DataType.DT_NTEXT:
                default:
                    return typeof(string);

            }
        }

        private void AddProperty(string name, string description, object value, bool supportsExpression)
        {

            IDTSCustomProperty100 commandProp = this.ComponentMetaData.CustomPropertyCollection.New();
            commandProp.Name = name;
            commandProp.Description = description;
            if (supportsExpression)
            {
                commandProp.ExpressionType = DTSCustomPropertyExpressionType.CPET_NOTIFY;
            }
            commandProp.Value = value;
            return;
        }

        private void FireEvent(EventType eventType, string description)
        {
            bool cancel = false;

            switch (eventType)
            {
                case EventType.Information:
                    this.ComponentMetaData.FireInformation(0, this.ComponentMetaData.Name, description, string.Empty, 0, ref cancel);
                    break;
                case EventType.Progress:
                    throw new NotImplementedException("Progress messages are not implemented");
                case EventType.Warning:
                    this.ComponentMetaData.FireWarning(0, this.ComponentMetaData.Name, description, string.Empty, 0);
                    break;
                case EventType.Error:
                    this.ComponentMetaData.FireError(0, this.ComponentMetaData.Name, description, string.Empty, 0, out cancel);
                    break;
                default:
                    this.ComponentMetaData.FireError(0, this.ComponentMetaData.Name, description, string.Empty, 0, out cancel);
                    break;
            }
        }

        #endregion

    }

    static class Constants
    {
        public const string CONNECTION_MANAGER_NAME = "BatchConnectionManager";
        public const string PROP_SQL_COMMAND = "SQLCommand";
        public const string PROP_WORKING_TABLE = "WorkingTable";

        public const uint DTS_PIPELINE_CTR_ROWSREAD = 101;
        public const uint DTS_PIPELINE_CTR_ROWSWRITTEN = 103;
        public const uint DTS_PIPELINE_CTR_BLOBBYTESREAD = 116;
        public const uint DTS_PIPELINE_CTR_BLOBBYTESWRITTEN = 118;
    }

    enum EventType
    {
        Information = 0,
        Progress = 1,
        Warning = 2,
        Error = 3
    }

    class ColumnInfo
    {
        private string _columnName = string.Empty;
        private DataType _dataType = DataType.DT_STR;
        private int _bufferIndex = 0;
        private int _precision = 0;
        private int _scale = 0;
        private int _length;

        public ColumnInfo(string columnName, DataType dataType, int bufferIndex, int length, int precision, int scale)
        {
            _columnName = columnName;
            _dataType = dataType;
            _bufferIndex = bufferIndex;
            _precision = precision;
            _scale = scale;
            _length = length;
        }

        public int BufferIndex
        { get { return _bufferIndex; } }

        public DataType ColumnDataType
        { get { return _dataType; } }

        public string ColumnName
        { get { return _columnName; } }

        public int Precision
        { get { return _precision; } }

        public int Length
        { get { return _length; } }

        public int Scale
        { get { return _scale; } }
    }
}
